package MariposaDB::Gossip;

use strict;

our @ISA = qw/Object::Event/; # gives us event support                                                                                                                                      

use Object::Event;

use Log::Log4perl qw(:easy);
use Data::Dumper;
use Coro;
use Coro::Signal;
use Cache::Memcached::Fast;


use MariposaDB::Data::Messages;
use MariposaDB::Client;

sub new
{   
    my $class = shift;
    my ($self)  = @_;

    $self = $self || {};
    bless ($self, $class);

    # setup our memcached connection
    unless ($self->_memcached_connect())
    {
	die "Unable to connecto to memcached. Verify it is up.";
    }

    # and some signals
    $self->{'init_signal'} = new Coro::Signal;

    return $self;
}

sub our_nodes
{
    my $self = shift;

    my ($our_nodes) = @_;

    $self->{'our_nodes'} = $our_nodes;
}

sub _memcached_connect
{
    my $self = shift;
    
    my $memcached = $self->{'config'}->get('memcached') || "localhost:11211";

    DEBUG "Using memcached: $memcached";
    
    $self->{'memcached'} = new Cache::Memcached::Fast
	(
	 {
	     'servers' => 
		 [
		  {
		      'address' => $memcached,
		      'weight' => 1
		  }
		 ],
	     'namespace' => $self->{'physical_node'},
	     'nowait' => 1
	 }
	);

    my $memcached_version = Dumper $self->{'memcached'}->server_versions();
    DEBUG "memcached version: $memcached_version";

    return $self->{'memcached'};
}

sub get_callback_id
{
    my $self = shift;

    # just a unique id
    return ++$self->{'last_callback_id'};
}

sub init
{
    # on first startup, we initialize our roster by contacting the seed host to see what they have
    my $self = shift;

    # as this might take some time, we do this in a Coro async block. It should return immediately.
    # $self->{'init'} will be set when it's done.

    return if $self->{'init'}; # we've already done the init, only on startup?

  async
  {
      # in case we are the seed host, we first write our virtual node data to memcached
      my $gossip_roster = $self->create_gossip_roster_from_node_list($self->{'our_nodes'});
      DEBUG "Our Roster is: ".length($gossip_roster);

      cede; # Question: what's a good time to cede? is this too much?

      my $roster_digest = $self->store_roster($gossip_roster); # this is the packed roster

      cede;

      # then we contact the seed, getting its roster
      my $gossip_id = $self->get_callback_id();
      my $seed_roster = $self->gossip_with
	  (
	   $self->{'seed_host'}, 
	   {
	       'roster_digest' => $roster_digest,
	       'gossip_id' => $gossip_id # will fire this event when done.
	   }
	  );
        
      # and initialized will be set once that's done via this callback
      $self->reg_cb 
	  (
	   $gossip_id => sub
	   {
	       my (undef, $ok) = @_;

	       DEBUG "INIT Done? $ok";

	       if ($ok)
	       {
		   $self->{'init'} = 1;
		   $self->{'init_signal'}->send(); # wait for this on start
	       }
	   }
	  );
  };
}

sub gossip_with
{
    my $self = shift;

    # a monster.  Just making it work. Please refactor.

    my ($host_colon_port, $params) = @_;

    my ($host,$port) = split(':',$host_colon_port);
    DEBUG "host: $host port $port";
    DEBUG Dumper $params;
    my $client = new MariposaDB::Client(
	{
	    'host' => $host,
	    'port' => $port
	});

    $client->reg_cb(
	'on_connect',
	sub
	{
	    DEBUG "On connect in gossip";

	    my $Gossip = MariposaDB::Data::Messages::Gossip->new();
	    # set the digest only on the first request.
	    $Gossip->set_node_list_digest($params->{'roster_digest'});
	    
	    my $Messages = MariposaDB::Data::Messages->new();
	    # add it to the message
	    $Messages->set_gossip($Gossip);
	    
	    # send it to the client
	    my $packMessages = $Messages->pack();
	    DEBUG "Sending gossip request: '$packMessages'";

	    $client->add_message($packMessages);
	    DEBUG "Message queued, waiting for response...";

	    async
	    {
		# now wait for a response
		my $response;
		while (!$response)
		{
		    $client->{'new_incoming_signal'}->wait;
		    
		    # get the response, hopefully
		    $response = $client->get_message();
		}
		
		$Messages = MariposaDB::Data::Messages->new;
		my $response_gossip; #placeholder for a response
		
	      SETTLE_NODE_LIST:
		{
		    my $tries = 2;
		    while ($tries)
		    {
			$tries--;
			DEBUG "tries: $tries";
			if ( $Messages->unpack($response) ) 
			{
			    DEBUG "PARSE OK";
			    my $has_gossip = $Messages->has_gossip();
			    DEBUG "has gossip $has_gossip";

			    # compare with our current digest
			    $response_gossip = $Messages->gossip();
			    my $response_digest = $response_gossip->node_list_digest();
			    
			    # if digests are different or it wasn't given ask for the full node list if they didn't return it already
			    if ( ($response_digest ne $params->{'roster_digest'}) || (!($response_digest)) )
			    {
				# check to see if we have a full node_list already.
				if ($response_gossip->node_list_size())
				{
				    DEBUG "how about that, they gave us a node list.";
				    last SETTLE_NODE_LIST; # yes, we got something.
				}
				
				# we have to ask for the full thing.
			      ASK_FOR_FULL_ROSTER:
				{
				    DEBUG "asking for the full gossip";		    
				    $Gossip->set_full(1); # the original request
				    $Gossip->clear_node_list_digest();
				    $Messages = MariposaDB::Data::Messages->new;
				    $Messages->set_gossip($Gossip);
				    $packMessages = $Messages->pack();
				    $client->add_message($packMessages);
				}
				
				# now wait for a response
				$response = undef;
				while (!$response)
				{
				    $client->{'new_incoming_signal'}->wait;
				    # get the response, hopefully
				    $Messages = MariposaDB::Data::Messages->new;
				    $response = $client->get_message();
				}
				# this should cycle around again and extract out the node_list so we can move on.
				DEBUG "Trying again...";
			    }
			}
			else
			{
			    DEBUG "PARSE ERROR??";
			    return; #??
			}
		    }
		}
		
	      MERGE_AND_STORE:
		{
		    unless ($response_gossip)
		    {
			DEBUG "No response gossip";
			next;
		    }
		
		    # merge if needed
		    my $current_roster = $self->get_roster();
		    my $current_gossip = MariposaDB::Data::Messages::Gossip->new;
		    $current_gossip->unpack($current_roster->{'roster'});
		    
		    # manual merge - tired at the moment there must be a better way to do this?
		    my $current_roster_hash = $self->create_roster_hash($current_gossip);
		    my $peer_roster_hash = $self->create_roster_hash($response_gossip);
		    
		    my $all_node = {};
		    foreach my $node_id (keys %{ $current_roster_hash}, keys %{ $peer_roster_hash })
		    {
			next if ($all_node->{$node_id}); # we've already done this node
			DEBUG "node id $node_id";

			if ($current_roster_hash->{$node_id} && $peer_roster_hash->{$node_id})
			{
			    $all_node->{$node_id} = $current_roster_hash->{$node_id}->merge($peer_roster_hash->{$node_id});
			}
			elsif ($peer_roster_hash->{$node_id})
			{
			    $all_node->{$node_id} = $peer_roster_hash->{$node_id};
			}
			elsif ($current_roster_hash->{$node_id})
			{
			    $all_node->{$node_id} = $current_roster_hash->{$node_id};
			}
		    }
		    
		    # reset list
		    $current_gossip->clear_node_list();
		    foreach my $node_id (sort { $a cmp $b } keys %{ $all_node })
		    {
			# add back one at a time, on node_id order
			$current_gossip->add_node_list($node_id);
		    }
		    
		    $current_gossip->clear_full(); #clear request bits
		    $current_gossip->clear_node_list_digest(); # allow the digest to be computed on storage
		    
		    # and store
		    $self->store_roster($current_gossip->pack());
		}
	    
		DEBUG "done gossip'n, for now.";
		$self->event($params->{'gossip_id'},1); # done?
	    };
	}
	);

    DEBUG "Waiting for callbacks...";
}

sub create_roster_hash
{
    my $self = shift;

    my ($gossip) = @_;

    # hashify the gossip struct
    my $hash = {};

    foreach my $node ($gossip->node_list())
    {
	$hash->{$node->node_identity()} = $node;
    }

    return $hash;
}

sub store_roster
{
    my $self = shift;

    my ($gossip_roster) = @_;

    # We store both the full roster and a digest

    # compute this now so we don't have to later
    # make sure the digest/full fields are unset to compute this.
    my $Gossip = MariposaDB::Data::Messages::Gossip->new();
    $Gossip->unpack($gossip_roster);
    $Gossip->clear_full();
    $Gossip->clear_node_list_digest();
    $gossip_roster = $Gossip->pack();

    my $digest = Digest::SHA1::sha1_hex($gossip_roster);

    DEBUG "Digest now $digest\n";

    $self->{'memcached'}->set_multi(
	['roster',$gossip_roster],
	['roster_digest', $digest]
	);

    return $digest;
}

sub get_roster
{
    my $self = shift;
    my ($type) = @_;

    my (@hkey) = ($type);
    unless (@hkey)
    {
	@hkey = ('roster', 'roster_digest');
    }

    # this gets what we currently think our roster is.
    my $roster = $self->{'memcached'}->get_multi(@hkey);
    return $roster; # access via $roster->{'roster'} and $roster->{'roster_digest'};
}

sub create_gossip_roster_from_node_list
{
    my $self = shift;
    my ($node_list) = @_;

    return undef unless ($node_list && @{$node_list});

    my $Gossip = MariposaDB::Data::Messages::Gossip->new();

    foreach my $node (@{ $node_list })
    {
	my $Node = MariposaDB::Data::Messages::Node->new();
	
	# node_id - sha1 of physical:number. Stored as a hexstring. Not necessarily == position, although numerically close.
	$Node->set_node_id($node->{'node_id'});
	
	# position - position on ring of start of bucket.
	my $RingPosition = MariposaDB::Data::Messages::RingPosition->new();
	$RingPosition->set_position($node->{'position'}->as_hex()); # position is a Math::BigInt, as hex starts with 0x
	$Node->set_position($RingPosition);

	# bucket count
	$Node->set_bucket_count($node->{'buckets'});

	# node location from config
	my $NodeIdentity = MariposaDB::Data::Messages::NodeIdentity->new();
	# for now, just the physical node id. add in your own replication strategies.
	$NodeIdentity->set_instance_id($self->{'physical_node'});
	$Node->set_node_identity($NodeIdentity);

	$Gossip->add_node_list($Node);
    }

    my $packed_gossip = $Gossip->pack();

    return $packed_gossip;
}

sub start
{
    my $self = shift;

    DEBUG "Starting Gossip...";

    # initialize our roster via the seed host
    $self->init();
    DEBUG "Init in process.  Waiting...";

    while (!($self->{'init'}))
    {
	$self->{'init_signal'}->wait;
	DEBUG "Init now:".$self->{'init'};
    }

    # great.
    DEBUG "Starting normal gossip...";
    # normal operations
    # every N seconds, pick a random host, get their roster fingerprint, compare to ours, ask for the full if different, merge, store
}


1;
